package org.seed.quantum.protocol.udp;

import org.seed.quantum.socket.MessageProcessor;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;



public class UdpDispatcher implements Runnable {

    public final static RequestTask EXECUTE_TASK_OR_SHUTDOWN = new RequestTask(null, null);
    private final BlockingQueue<RequestTask> taskQueue = new LinkedBlockingQueue<>();
    private final MessageProcessor processor;

    public UdpDispatcher(MessageProcessor processor) {
        this.processor = processor;
    }

    @Override
    public void run() {
        while (true) {
            try {
                RequestTask unit = taskQueue.take();
                if (unit == EXECUTE_TASK_OR_SHUTDOWN) {
                    break;
                }
                processor.process(unit.session, unit.request);
                if (!unit.session.isInvalid()) {
                    unit.session.writeBuffer().flush();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 任务分发
     */
    public void dispatch(UdpSession session, Object request) {
        dispatch(new RequestTask(session, request));
    }

    /**
     * 任务分发
     */
    public void dispatch(RequestTask requestTask) {
        taskQueue.offer(requestTask);
    }

    static class RequestTask {
        UdpSession session;
        Object request;

        public RequestTask(UdpSession session, Object request) {
            this.session = session;
            this.request = request;
        }
    }
}
